package com.thp.bigdata.logEnhance;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.junit.Test;

/**
 * 日志增强：
 * 写入不同的文件
 * @author 汤小萌
 *
 */
public class LogEnhance {
	static class LogEnhanceMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		Map<String, String> urlContentMap = new HashMap<String, String>();
		Text k = new Text();
		NullWritable v = NullWritable.get();
		
		/**
		 * 从数据库中加载数据到HashMap缓存下来
		 */
		@Override
		protected void setup(Context context)
				throws IOException, InterruptedException {
			DBLoader.dbLoader(urlContentMap);
		}
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// 获取一个计数器 - (这个计数器是全局的计数器)  记录不合法的日志行数，组名，计数器名称
			Counter counter = context.getCounter("malFormed", "malFormedCounter");
			String line = value.toString();
			String[] fields = line.split("\t");
			try {
				String url = fields[28];
				System.out.println(url);
				String content_tag = urlContentMap.get(url);
				// System.out.println(content_tag);
				if(content_tag == null) { // 从知识库中根据对应的url查询的内容为空，
					k.set(url + "\t" + "tocrawl" + "\n");  // 自定义的输出流没有包装，不能换行 
					context.write(k, v);
				} else {
					k.set(line + "\t" + content_tag + "\n");
					System.out.println(k.toString());
					context.write(k, v);
				}
			} catch (Exception e) {
				// 有的数据可能是不合法的，长度不够，不完整的数据
				e.printStackTrace();
				System.err.println("数据不合法");
				counter.increment(1);
			}
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		
		
		job.setJarByClass(LogEnhance.class);
		job.setMapperClass(LogEnhanceMapper.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		// 要控制不同的文件内容写往不同的目标路径，采用自定义的OutputStream
		job.setOutputFormatClass(LogEnhanceOutputFormat.class);
		
		FileInputFormat.setInputPaths(job, new Path("f:/enhancelog/input"));
		
		// 尽管我们在自定义的OutputFormat里面已经设置好了输出的路径
		// 但是在FileOutputFormat中，必须输出一个_success文件，所以还需要设置输出path
		FileOutputFormat.setOutputPath(job, new Path("f:/enhancelog/output"));
		
		// 现在只是做日志的清洗，还不需要reduce task
		job.setNumReduceTasks(0);
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		
	}
	
	
	
	@Test
	public void test() {
		// String str = "1374609560.11	1374609560.16	1374609560.16	1374609560.16	110	5	8615038208365	460023383869133	8696420056841778	2	460	0	14615			54941	10.188.77.252	61.145.116.27	35020	80	6	cmnet	1	221.177.218.34	221.177.217.161	221.177.218.34	221.177.217.167	ad.veegao.com	http://ad.veegao.com/veegao/iris.action		Apache-HttpClient/UNAVAILABLE (java 1.4)	POST	200	593	310	4	3	0	0	4	3	0	0	0	0	http://ad.veegao.com/veegao/iris.action	5903903079251243019	5903903103500771339	5980728";
		String str = "1374609557.12	1374609557.15	1374609557.15	1374609557.74	110	5	8615093268715	460023934411519	3588660433773101	2	460	0	14822			29343	10.188.77.164	223.203.194.156	42384	80	6	cmnet	1	221.177.218.41	221.177.217.161	221.177.218.41	221.177.217.167	ugc.moji001.com	http://ugc.moji001.com/sns/GetNewestShare/100/489?UserID=42958568&Platform=Android&Version=10023802&BaseOSVer=10&PartnerKey=5007&Model=GT-S7500&Device=phone&VersionType=1&TS=		Apache-HttpClient/UNAVAILABLE (java 1.4)	GET	200	421	363	3	2	0	0	3	2	0	0	0	0	http://ugc.moji001.com/sns/GetNewestShare/100/489?UserID=42958568&Platform=Android&Version=10023802&BaseOSVer=10&PartnerKey=5007&Model=GT-S7500&Device=phone&VersionType=1&TS=	5903903047315243019	5903903087191863307	5980488";
		
		String[] fields = str.split("\t");
		int count = 1;
		for(String field : fields) {
			System.out.println(count + " >   " + field);
			count++;
		}
	}
	
	
	
}
